"""
@Author: CheyenneLam
@Date: 2020-08-01 09:00:00
@LastEditTime: 2020-08-01 09:00:00
@LastEditors: CheyenneLam
@Description: 生成增量表、全量表的全增量合并SQL语句
@FilePath: \create_merge_data.py
"""

from odps.df import DataFrame

# 指定表
# 如果tableList为空的话, 则查询整个system_table_info
tableList = [
    'common.coupon_template'
]

# 获取表名、主键
df = DataFrame(o.get_table('system_table_info')).to_pandas().query("data_base == 'pg'")
if len(tableList) > 0:
    pk_dict = df.query("data_source_table in @tableList")[['dw_full_table_name', 'pk_name']]
else:
    pk_dict = df[['dw_full_table_name', 'pk_name']]

insert_sql_collection = []

for table_name, ids in pk_dict.values:
    inc_table = table_name.replace("full_", "inc_")
    binlog_table = f"{table_name}_binlog"
    partition_id_sql = ' , '.join([f'''{key}''' for key in ids])
    df = o.get_table(table_name).to_df()

    with_del_sql = f"""
        WITH DEL AS (
            select {partition_id_sql} from (
                SELECT ROW_NUMBER() OVER (PARTITION BY {partition_id_sql} 
                    ORDER BY t.sequence_id DESC,t.after_image DESC) AS nums
                    ,t.*
                FROM {binlog_table} t 
                where t.mdf_year = '${{mdf_year}}' and t.mdf_month='${{mdf_month}}' and t.mdf_day='${{mdf_day}}'
            ) t_outer
            where t_outer.nums = 1
            and t_outer.operation_type = 'DELETE'
        )
    """

    id_status_sql = ' and '.join([f'''inc.{id} IS NOT NULL''' for id in ids])

    columns = ''
    hasComplicatedType = False
    for x in df.schema:
        column = f'''CASE WHEN {id_status_sql} THEN inc.{x.name} ELSE t.{x.name} end as {x.name}'''
        if x.type.name.find('list') != -1 or x.type.name.find('map') != -1:
            hasComplicatedType = True

        elif x.type.name.strip() == 'datetime':
            column = f'''CASE WHEN {id_status_sql} 
            THEN IF(ISDATE(inc.{x.name}, 'yyyy-mm-dd hh:mi:ss'), TO_DATE(inc.{x.name},'yyyy-mm-dd hh:mi:ss'),NULL) 
            ELSE IF(ISDATE(t.{x.name}, 'yyyy-mm-dd hh:mi:ss'), TO_DATE(t.{x.name},'yyyy-mm-dd hh:mi:ss'),NULL)  end as {x.name}
            '''

        columns += column
        if df.schema.columns[len(df.schema.columns) - 1].name != x.name:
            columns += '\n,'
    # print(columns)
    # columns = ',\t \n'.join(
    #     [f'''CASE WHEN {id_status_sql} THEN inc.{x.name} ELSE t.{x.name} end as {x.name}''' for x in df.schema])
    join_sql = ' and '.join([f'''inc.{id} = t.{id}''' for id in ids])
    outer_join_sql = ' and '.join([f'''TEMP.{id} = DEL.{id}''' for id in ids])

    insert_sql = f"""
                \n
                {with_del_sql} 
                INSERT OVERWRITE TABLE {table_name}
                    SELECT TEMP.* FROM (
                    SELECT {"" if hasComplicatedType else "DISTINCT"} {columns}
                FROM {table_name} AS t 
                FULL OUTER JOIN {inc_table} AS inc
                ON {join_sql}
                ) TEMP LEFT ANTI JOIN DEL
                ON {outer_join_sql};

                """
    print(f"""
    .
    .
    .
    """)
    print(insert_sql)